package org.quincy.rock.core.concurrent;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;

import org.quincy.rock.core.exception.BlcokingException;
import org.quincy.rock.core.exception.RockException;
import org.quincy.rock.core.util.CoreUtil;
import org.quincy.rock.core.util.DateUtil;

/**
 * <b>使用阻塞队列缓存的数据处理服务。</b>
 * <p><b>详细说明：</b></p>
 * <!-- 在此添加详细说明 -->
 * 无。
 * <p><b>修改列表：</b></p>
 * <table width="100%" cellSpacing=1 cellPadding=3 border=1>
 * <tr bgcolor="#CCCCFF"><td>序号</td><td>作者</td><td>修改日期</td><td>修改内容</td></tr>
 * <!-- 在此添加修改列表，参考第一行内容 -->
 * <tr><td>1</td><td>wks</td><td>2018年5月25日 下午3:56:41</td><td>建立类型</td></tr>
 * 
 * </table>
 * @version 1.0
 * @author wks
 * @since 1.0
 */
@SuppressWarnings({ "unchecked", "rawtypes" })
public abstract class QueueProcessService<K, V> extends ProcessService<K, V> {
	/**
	 * 阻塞超时秒数。
	 */
	private int timeout = 30;
	/**
	 * 多个线程共享一个队列。
	 */
	private boolean sharedQueue;
	/**
	 * 线程组。
	 */
	private final ThreadGroup threadGroup;
	/**
	 * 线程数。
	 */
	private final int threadCount;
	/**
	 * 执行者线程组。
	 */
	private Executor<DataClosure<K, V>>[] executors;

	/**
	 * <b>构造方法。</b>
	 * <p><b>详细说明：</b></p>
	 * <!-- 在此添加详细说明 -->
	 * 无。
	 * @param tc 线程数
	 * @param sharedQueue 多个线程共享一个队列
	 */
	public QueueProcessService(int tc, boolean sharedQueue) {
		this.threadCount = Math.max(tc, 1);
		this.sharedQueue = sharedQueue;
		threadGroup = new ThreadGroup(getId());
		threadGroup.setMaxPriority(Thread.MAX_PRIORITY);
	}

	/**
	 * <b>初始化执行者线程。</b>
	 * <p><b>详细说明：</b></p>
	 * <!-- 在此添加详细说明 -->
	 * 无。
	 */
	private synchronized void initExecutorThread() {
		if (executors == null) {
			executors = new Executor[threadCount];
			BlockingQueue<DataClosure<K, V>> queue = sharedQueue ? createBlockingQueue() : null;
			for (int i = 0; i < threadCount; i++) {
				executors[i] = new Executor(i, queue == null ? createBlockingQueue() : queue, threadGroup,
						getId() + "_" + i);
				executors[i].setPriority(Thread.MAX_PRIORITY);
				executors[i].start();
			}
		}
	}

	/** 
	 * start。
	 * @see org.quincy.rock.core.concurrent.ProcessService#start()
	 */
	@Override
	public void start() {
		if (executors == null)
			initExecutorThread();
		super.start();
	}

	/**
	 * <b>创建阻塞队列。</b>
	 * <p><b>详细说明：</b></p>
	 * <!-- 在此添加详细说明 -->
	 * 无。
	 * @return 阻塞队列
	 */
	protected abstract BlockingQueue<DataClosure<K, V>> createBlockingQueue();

	/**
	 * <b>返回当前运行线程个数。</b>
	 * <p><b>详细说明：</b></p>
	 * <!-- 在此添加详细说明 -->
	 * 无。
	 * @return 当前运行线程个数
	 */
	public final int activeCount() {
		return threadGroup.activeCount();
	}

	/**
	 * <b>获得线程数。</b>
	 * <p><b>详细说明：</b></p>
	 * <!-- 在此添加详细说明 -->
	 * 无。
	 * @return 线程数
	 */
	public final int getThreadCount() {
		return this.threadCount;
	}

	/**
	 * 返回阻塞超时秒数
	 * @return 阻塞超时秒数
	 */
	public final int getTimeout() {
		return timeout;
	}

	/**
	 * 设置阻塞超时秒数
	 * @param timeout 阻塞超时秒数
	 */
	public final void setTimeout(int timeout) {
		this.timeout = timeout;
	}

	/**
	 * <b>多个线程共享一个队列。</b>
	 * <p><b>详细说明：</b></p>
	 * <!-- 在此添加详细说明 -->
	 * 无。
	 * @return 多个线程共享一个队列
	 */
	public final boolean isSharedQueue() {
		return sharedQueue;
	}

	/**
	 * <b>返回队列是否为空。</b>
	 * <p><b>详细说明：</b></p>
	 * <!-- 在此添加详细说明 -->
	 * 无。
	 * @return 返回队列是否为空
	 */
	public final boolean isEmpty() {
		if (executors == null)
			return true;
		if (isSharedQueue()) {
			int len = getThreadCount();
			for (int i = 0; i < len; i++) {
				Executor<DataClosure<K, V>> executor = executors[i];
				if (executor != null)
					return executor.isEmpty();
			}
		} else {
			int len = getThreadCount();
			for (int i = 0; i < len; i++) {
				if (count(i) > 0)
					return false;
			}
		}
		return true;
	}

	/**
	 * <b>等待队列中所有数据处理完毕。</b>
	 * <p><b>详细说明：</b></p>
	 * <!-- 在此添加详细说明 -->
	 * 无。
	 */
	public final void waitAllDone() {
		while (!(isEmpty() && isIdle())) {
			DateUtil.sleep(500);
		}
	}

	/**
	 * <b>返回所有线程是否是空闲的。</b>
	 * <p><b>详细说明：</b></p>
	 * <!-- 在此添加详细说明 -->
	 * 无。
	 * @return 所有线程是否是空闲的
	 */
	public final boolean isIdle() {
		if (executors != null) {
			for (int i = 0, len = getThreadCount(); i < len; i++) {
				if (!isIdle(i))
					return false;
			}
		}
		return true;
	}

	/**
	 * <b>返回指定线程是否是空闲的。</b>
	 * <p><b>详细说明：</b></p>
	 * <!-- 在此添加详细说明 -->
	 * 无。
	 * @param threadIndex 线程索引
	 * @return 指定线程是否是空闲的
	 */
	public final boolean isIdle(int threadIndex) {
		if (executors == null)
			return true;
		Executor<DataClosure<K, V>> executor = executors[threadIndex];
		return executor == null ? true : executor.isIdle();
	}

	/**
	 * <b>返回指定线程的待处理数据条数。</b>
	 * <p><b>详细说明：</b></p>
	 * <!-- 在此添加详细说明 -->
	 * 无。
	 * @param threadindex 线程索引
	 * @return 队列中待处理数据条数
	 */
	public final int count(int threadIndex) {
		if (executors == null)
			return 0;
		Executor<DataClosure<K, V>> executor = executors[threadIndex];
		return executor == null ? 0 : executor.count();
	}

	/**
	 * <b>返回待处理总条数。</b>
	 * <p><b>详细说明：</b></p>
	 * <!-- 在此添加详细说明 -->
	 * 无。
	 * @return 待处理总条数
	 */
	public final int count() {
		if (executors == null)
			return 0;
		if (isSharedQueue()) {
			int len = getThreadCount();
			for (int i = 0; i < len; i++) {
				Executor<DataClosure<K, V>> executor = executors[i];
				if (executor != null)
					return executor.count();
			}
			return 0;
		} else {
			int len = getThreadCount();
			int total = 0;
			for (int i = 0; i < len; i++) {
				total += count(i);
			}
			return total;
		}
	}

	/**
	 * <b>>返回指定线程的处理数据条数。</b>
	 * <p><b>详细说明：</b></p>
	 * <!-- 在此添加详细说明 -->
	 * 无。
	 * @param threadIndex 线程索引
	 * @return 指定线程的处理数据条数
	 */
	public final long processCount(int threadIndex) {
		if (executors == null)
			return 0;
		Executor<DataClosure<K, V>> executor = executors[threadIndex];
		return (executor == null) ? 0 : executor.processCount();
	}

	/**
	 * <b>返回处理数据总条数。</b>
	 * <p><b>详细说明：</b></p>
	 * <!-- 在此添加详细说明 -->
	 * 无。
	 * @return 处理数据总条数
	 */
	public final long processCount() {
		int len = getThreadCount();
		long total = 0;
		for (int i = 0; i < len; i++) {
			total += processCount(i);
		}
		return total;
	}

	/**
	 * <b>>返回指定线程的单条数据处理毫秒数。</b>
	 * <p><b>详细说明：</b></p>
	 * <!-- 在此添加详细说明 -->
	 * 无。
	 * @param threadIndex 线程索引
	 * @return 指定线程的单条数据处理毫秒数
	 */
	public final long processMillis(int threadIndex) {
		if (executors == null)
			return 0;
		Executor<DataClosure<K, V>> executor = executors[threadIndex];
		return (executor == null) ? 0 : executor.processMillis();
	}

	/**
	 * <b>返回单条数据平均处理毫秒数。</b>
	 * <p><b>详细说明：</b></p>
	 * <!-- 在此添加详细说明 -->
	 * 无。
	 * @return 单条数据平均处理毫秒数
	 */
	public final long processMillis() {
		int len = getThreadCount();
		long total = 0;
		for (int i = 0; i < len; i++) {
			total += processMillis(i);
		}
		return total / len;
	}

	/**
	 * <b>返回指定线程的数据等待秒数。</b>
	 * <p><b>详细说明：</b></p>
	 * <!-- 在此添加详细说明 -->
	 * 指示数据在队列中等待多久会被处理,反映线程的处理效率。
	 * @return 指定线程的数据等待秒数
	 */
	public final int waitSeconds(int threadIndex) {
		if (executors == null)
			return 0;
		Executor<DataClosure<K, V>> executor = executors[threadIndex];
		return executor == null ? 0 : executor.waitSeconds();
	}

	/**
	 * <b>返回数据平均等待秒数。</b>
	 * <p><b>详细说明：</b></p>
	 * <!-- 在此添加详细说明 -->
	 * 指示数据在队列中等待多久会被处理,反映线程的处理效率。
	 * @return 数据平均等待秒数
	 */
	public final int waitSeconds() {
		int len = getThreadCount();
		int total = 0;
		for (int i = 0; i < len; i++) {
			total += waitSeconds(i);
		}
		return total / len;
	}

	/** 
	 * handleDataClosure。
	 * @see org.quincy.rock.core.concurrent.ProcessService#handleDataClosure(org.quincy.rock.core.concurrent.ProcessService.DataClosure)
	 */
	@Override
	protected final void handleDataClosure(DataClosure<K, V> dataClosure) throws BlcokingException {
		int threadIndex = assignExecutorIndex(dataClosure.key(), getThreadCount());
		executors[threadIndex].put(dataClosure);
	}

	/**
	 * <b>分配线程执行者索引。</b>
	 * <p><b>详细说明：</b></p>
	 * <!-- 在此添加详细说明 -->
	 * 无。
	 * @param key 键
	 * @param totalCount 总共线程数
	 * @return 线程执行者索引
	 */
	protected int assignExecutorIndex(K key, int totalCount) {
		return easiest(totalCount);
	}

	/**
	 * <b>获得最轻松的线程索引。</b>
	 * <p><b>详细说明：</b></p>
	 * <!-- 在此添加详细说明 -->
	 * 无。
	 * @param totalCount 总共线程数
	 * @return 最轻松的线程索引
	 */
	protected final int easiest(int totalCount) {
		if (isSharedQueue())
			return 0;
		else {
			int minCount = count(0), minIndex = 0;
			if (minCount != 0) {
				for (int index = 1; index < totalCount; index++) {
					int count = count(index);
					if (count == 0) {
						minIndex = index;
						break;
					} else if (minCount > count) {
						minCount = count;
						minIndex = index;
					}
				}
			}
			return minIndex;
		}
	}

	/** 
	 * awaitTermination。
	 * @see org.quincy.rock.core.concurrent.ProcessService#awaitTermination(long)
	 */
	@Override
	protected void awaitTermination(long timeout) throws Exception {
		if (executors == null)
			return;
		int len = executors.length;
		//设置停止标记
		for (int i = 0; i < len; i++) {
			Executor exec = executors[i];
			if (exec != null)
				executors[i].setStopFlag();
		}
		//join
		if (timeout > 0) {
			timeout *= 1000;
			long base = System.currentTimeMillis();
			for (int i = 0; i < len; i++) {
				Executor exec = executors[i];
				if (exec != null) {
					exec.join(timeout);
					timeout = timeout - (System.currentTimeMillis() - base);
					if (timeout > 0)
						base = System.currentTimeMillis();
					else
						break; //没有剩余超时了。
				}
			}
		} else {
			for (int i = 0; i < len; i++) {
				Executor exec = executors[i];
				if (exec != null)
					exec.join(timeout);
			}
		}
		//置空
		for (int i = 0; i < len; i++) {
			CoreUtil.interrupt(executors[i]);
			executors[i] = null;
		}
		executors = null;
		CoreUtil.interrupt(threadGroup);
		threadGroup.destroy();
	}

	//执行者
	protected class Executor<T extends DataClosure<K, V>> extends Thread {
		/**
		 * threadIndex。
		 */
		private final int threadIndex;
		/**
		 * queue。
		 */
		private BlockingQueue<T> queue;
		/**
		 * 线程停止标记。
		 */
		private boolean stopFlag;
		/**
		 * 数据等待秒数。
		 */
		private int waitSeconds;
		/**
		 * 单条数据处理毫秒数。
		 */
		private long processMillis;
		/**
		 * 处理数据条数。
		 */
		private long processCount;
		/**
		 * 空闲。
		 */
		private boolean idle;
		/**
		 * 最近一次工作时间。
		 */
		private long lastWorkTime;

		/**
		 * <b>构造方法。</b>
		 * <p><b>详细说明：</b></p>
		 * <!-- 在此添加详细说明 -->
		 * 无。
		 * @param index 所在索引
		 * @param queue 队列
		 * @param group 线程组
		 * @param name 线程名称
		 */
		public Executor(int index, BlockingQueue<T> queue, ThreadGroup group, String name) {
			super(group, name);
			this.threadIndex = index;
			this.queue = queue;
		}

		/**
		 * <b>返回队列。</b>
		 * <p><b>详细说明：</b></p>
		 * <!-- 在此添加详细说明 -->
		 * 无。
		 * @return 队列
		 */
		private BlockingQueue<T> queue() {
			return queue;
		}

		/** 
		 * run。
		 * @see java.lang.Thread#run()
		 */
		@Override
		public void run() {
			while (true) {
				try {
					this.idle = true;
					DataClosure<K, V> closure = queue().poll(500, TimeUnit.MILLISECONDS);
					if (closure != null) {
						this.idle = false;
						long ibegin = System.currentTimeMillis();
						this.waitSeconds = (int) (ibegin - closure.timestamp()) / 1000;
						processDataClosure(closure);
						long iEnd = System.currentTimeMillis();
						this.processMillis = iEnd - ibegin;
						this.processCount++;
						this.lastWorkTime = iEnd;
						this.idle = true;
					}
					if (stopFlag && queue().isEmpty()) {
						//退出并销毁线程
						recorder.write("QueueProcessService[{0}]:Quit the queue[{1}].", getName(), threadIndex);
						break;
					}
				} catch (Exception e) {
					recorder.write(e, e.getMessage());
				}
			}
			this.idle = true;
			Executor[] execs = executors;
			if (execs != null)
				execs[threadIndex] = null;
		}

		/**
		 * <b>放入一条数据。</b>
		 * <p><b>详细说明：</b></p>
		 * <!-- 在此添加详细说明 -->
		 * 无。
		 * @param closure 数据
		 */
		public void put(T closure) {
			try {
				if (stopFlag || timeout > 0 ? !queue().offer(closure, timeout, TimeUnit.SECONDS)
						: !queue().offer(closure))
					throw new BlcokingException();
			} catch (InterruptedException e) {
				throw new RockException(e.getMessage(), e);
			}
		}

		/**
		 * <b>返回队列中待处理的数据个数。</b>
		 * <p><b>详细说明：</b></p>
		 * <!-- 在此添加详细说明 -->
		 * 无。
		 * @return 队列中待处理的数据个数
		 */
		public int count() {
			return queue().size();
		}

		/**
		 * <b>返回队列是否为空。</b>
		 * <p><b>详细说明：</b></p>
		 * <!-- 在此添加详细说明 -->
		 * 无。
		 * @return 返回队列是否为空
		 */
		public boolean isEmpty() {
			return queue().isEmpty();
		}

		/**
		 * <b>线程是空闲的。</b>
		 * <p><b>详细说明：</b></p>
		 * <!-- 在此添加详细说明 -->
		 * 无。
		 * @return 线程是空闲的
		 */
		public boolean isIdle() {
			return idle && (System.currentTimeMillis() - lastWorkTime) > 1500;
		}

		/**
		 * <b>数据等待秒数。</b>
		 * <p><b>详细说明：</b></p>
		 * <!-- 在此添加详细说明 -->
		 * 指示数据在队列中等待多久会被处理。
		 * @return 数据等待秒数
		 */
		public int waitSeconds() {
			return this.waitSeconds;
		}

		/**
		 * <b>单条数据处理毫秒数。</b>
		 * <p><b>详细说明：</b></p>
		 * <!-- 在此添加详细说明 -->
		 * 无。
		 * @return 单条数据处理毫秒数
		 */
		public long processMillis() {
			return this.processMillis;
		}

		/**
		 * <b>处理数据条数。</b>
		 * <p><b>详细说明：</b></p>
		 * <!-- 在此添加详细说明 -->
		 * 无。
		 * @return 处理数据条数
		 */
		public long processCount() {
			return this.processCount;
		}

		/**
		 * <b>设置停止标志。</b>
		 * <p><b>详细说明：</b></p>
		 * <!-- 在此添加详细说明 -->
		 * 通知停止执行者线程。
		 */
		public void setStopFlag() {
			stopFlag = true;
		}
	}
}
